package cn.com.rpc.dispatcher;

import cn.com.common.Constant;
import cn.com.common.URI;
import cn.com.rpc.HandlerException;
import cn.com.rpc.NettyChannel;
import cn.com.rpc.event.ChannelEventHandler;
import cn.com.rpc.handler.Handler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleStateEvent;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author jiaming
 */
public class AllHandlerDispatcher extends ChannelDuplexHandler {

    protected static final Logger log = LogManager.getLogger();

    private URI uri;
    private Handler handler;
    private static final ExecutorService executorService = new ThreadPoolExecutor(Constant.WORKER_THREAD_DEFAULT_NUM, Constant.WORKER_THREAD_DEFAULT_NUM, 5000,
            TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(8192), new ThreadFactory() {

        private AtomicInteger atomicInteger = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("all-handler-dispatcher-" + atomicInteger.incrementAndGet());
            return thread;
        }
    });


    public AllHandlerDispatcher(Handler handler, URI uri) {
        this.handler = handler;
        this.uri = uri;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        try {
            Channel channel = ctx.channel();
            NettyChannel nettyChannel = NettyChannel.getChannel(channel, uri);
            executorService.execute(new ChannelEventHandler(nettyChannel, handler, null, ChannelEventHandler.ChannelEventState.CONNECTED));
        } catch (Exception e) {
            throw new HandlerException(e.getMessage(), e);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.channel());
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        try {
            Channel channel = ctx.channel();
            NettyChannel nettyChannel = NettyChannel.getChannel(channel, uri);
            executorService.execute(new ChannelEventHandler(nettyChannel, handler, null, ChannelEventHandler.ChannelEventState.DISCONNECTED));
        } catch (Exception e) {
            throw new HandlerException(e.getMessage(), e);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.channel());
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object obj) throws Exception {
        try {
            Channel channel = ctx.channel();
            NettyChannel nettyChannel = NettyChannel.getChannel(channel, uri);
            executorService.execute(new ChannelEventHandler(nettyChannel, handler, obj, ChannelEventHandler.ChannelEventState.RECEIVED));
        } catch (Exception e) {
            throw new HandlerException(e.getMessage(), e);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.channel());
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            NettyChannel channel = NettyChannel.getChannel(ctx.channel(), uri);
            try {
                log.info("IdleStateEvent triggered, close channel " + channel);
                channel.close();
            } finally {
                NettyChannel.removeChannelIfDisconnected(ctx.channel());
            }
        }
        super.userEventTriggered(ctx, evt);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        try {
            Channel channel = ctx.channel();
            NettyChannel nettyChannel = NettyChannel.getChannel(channel, uri);
            executorService.execute(new ChannelEventHandler(nettyChannel, handler, null, ChannelEventHandler.ChannelEventState.CAUGHT, cause));
        } catch (Exception e) {
            throw new HandlerException(e.getMessage(), e);
        } finally {
            NettyChannel.removeChannelIfDisconnected(ctx.channel());
        }
    }
}
